package bigdata.hourglass.engine

import java.util.concurrent.locks.ReentrantLock
import java.util.concurrent.{ConcurrentHashMap, TimeUnit}

import bigdata.hourglass.common.domain.Bean.HourglassExecution
import org.apache.spark.sql.Row
import org.apache.spark.sql.streaming.{DataStreamWriter, StreamingQuery}

import scala.collection.JavaConverters._
import scala.collection.concurrent

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @organization DataReal
 * @version 1.0
 * @website https://www.jlpyyf.com
 * @date 2019/12/8 8:58 下午
 * @since 1.0
 */
class HourglassSession(_engineInfo: String, _tag: String) {
  def engineInfo: String = _engineInfo

  def tag: String = _tag

  // 保存batch任务engineInfo_iqlId和result的映射（获取某个SQL的结果或者kill，与对应engine通讯即可）
  val batchJob: ConcurrentHashMap[String, HourglassExecution] =
    new ConcurrentHashMap[String, HourglassExecution]()

  // 保存stream任务engineInfo和StreamingQuery的映射（查看某个stream的状态或者stop某个stream）
  val streamJob: concurrent.Map[String, StreamingQuery] =
    new ConcurrentHashMap[String, StreamingQuery]().asScala
  // 实时任务对应的邮件receiver
  val streamJobWithMailReceiver = new ConcurrentHashMap[String, String]()
  var streamJobWithDingDingReceiver: Set[String] = Set()
  val streamJobMaxAttempts = new ConcurrentHashMap[String, Int]()
  val streamJobWithDataFrame = new ConcurrentHashMap[String, DataStreamWriter[Row]]()

  private val lock = new ReentrantLock()
  private val condition = lock.newCondition()
  private var stopped: Boolean = false
  private var error: Throwable = _

  def awaitTermination(timeout: Long = -1): Boolean = {
    lock.lock()
    try {
      if (timeout < 0) {
        while (!stopped && error == null) {
          condition.await()
        }
      } else {
        var nanos = TimeUnit.MILLISECONDS.toNanos(timeout)
        while (!stopped && error == null && nanos > 0) {
          nanos = condition.awaitNanos(nanos)
        }
      }
      // If already had error, then throw it
      if (error != null) throw error
      // already stopped or timeout
      stopped
    } finally {
      lock.unlock()
    }
  }
}

object HourglassSession {

}
